package com.bianmaba.jobs.collection.monitor;

import com.alibaba.fastjson.JSON;
import com.bianmaba.beans.datacollection.CollectionTask;
import com.bianmaba.beans.datacollection.ExecuteLog;
import com.bianmaba.beans.datacollection.TaskInstance;
import com.bianmaba.configuration.MetricPropertiesConfiguration;
import com.bianmaba.dc.bean.CollectorResponse;
import com.bianmaba.dc.bean.JobParameter;
import com.bianmaba.dc.bean.RedisKeys;
import com.bianmaba.dc.bean.TaskStatus;
import com.bianmaba.remote.supports.CollectorRemoteRequestService;
import com.bianmaba.services.datacollection.ExecuteLogRecorder;
import com.bianmaba.services.datacollection.ITaskInstanceService;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;

import java.util.Calendar;
import java.util.Date;
import java.util.Map;

@Log4j2
public class CollectorStatusMonitor implements Job {
    @Autowired
    protected CollectorRemoteRequestService remoteRequestService;
    @Autowired
    protected ITaskInstanceService taskInstanceService;
    @Autowired
    protected ExecuteLogRecorder logWriter;
    @Autowired
    protected ListOperations<String, Object> redisListOperations;
    @Autowired
    protected MetricPropertiesConfiguration.MetricProperties metricProperties;

    private Date startCheckTime;

    private CollectionTask task;

    @Override
    public void execute(JobExecutionContext context) {
        JobDataMap dataMap = context.getMergedJobDataMap();
        String taskInstanceId = dataMap.getString(JobParameter.TASK_INSTANCE_ID.getName());
        TaskInstance taskInstance = taskInstanceService.findOne(taskInstanceId);

        logWriter.put(ExecuteLog.of(taskInstanceId, "任务状态监听启动"));
        task = taskInstance.getCollectionTask();

        try {
            checkStatus(taskInstanceId);
        } catch (InterruptedException e) {
            logWriter.put(ExecuteLog.of(taskInstanceId, "用户终止"));
        }
    }


    private void checkStatus(String taskInstanceId) throws InterruptedException {
        startCheckTime = new Date();
        Integer status = taskInstanceService.getStatus(taskInstanceId);
        while (true) {
            Thread.sleep(1000);
            //发送状态检查请求
            CollectorResponse result = sendStatusCheck(taskInstanceId);
            //处理执行日志
            handlExecutorLog(taskInstanceId, result);

            //返回success为true时，表示采集器正常返回（但状态可能不同）
            // 为false时表示请求出现错误，如地址，权限或其它问题，此时继续查询
            if (!result.isSuccess()) {
                //如果连续一小时都无法正常访问状态检查接口则终止任务
                if ((startCheckTime.getTime() + 1 * 60 * 60 * 1000) < System.currentTimeMillis()) {
                    taskInstanceService.updateStatus(taskInstanceId, TaskStatus.SYSTEM_TERMINATED.getIndex());
                    String msg = "由于长时间（超过一小时）无法访问采集器，任务终止。";
                    logWriter.put(ExecuteLog.of(taskInstanceId, msg));
                    log.warn(msg);
                    break;
                }
                continue;
            }
            //如果状态未更新，则退出当前循环
            Integer newStatus = result.getTaskStatus().getIndex();
            if (newStatus.equals(status) && TaskStatus.RUNNING.getIndex() == newStatus.byteValue()) {
                continue;
            }

            //如果状态未更新则刷新数据库，同时处理相关数据
            status = newStatus;

            if (TaskStatus.RUNNING.getIndex() == newStatus.byteValue()) {
                taskInstanceService.updateStatus(taskInstanceId, newStatus);
                continue;
            } else if (TaskStatus.SUCCESSFUL.getIndex() == newStatus.byteValue()) {
                String msg = "任务状态监听结束，任务状态：" + result.getTaskStatus();
                log.info(msg);
                logWriter.put(ExecuteLog.of(taskInstanceId, msg));
                Object datas = result.getData();
                if (datas != null && StringUtils.isNotEmpty(datas.toString())) {
                    msg = "采集器成功返回数据";
                    log.info(msg);
                    logWriter.put(ExecuteLog.of(taskInstanceId, msg));
                    redisListOperations.leftPush(RedisKeys.PENDING_DATA, JSON.toJSONString(datas));
                    msg = "将采集结果推入待处理队列中。";
                    log.info(msg);
                    logWriter.put(ExecuteLog.of(taskInstanceId, msg));
                } else {
                    msg = "采集器未返回数据";
                    log.info(msg);
                    logWriter.put(ExecuteLog.of(taskInstanceId, msg));
                }
                break;
            } else {
                taskInstanceService.updateStatus(taskInstanceId, newStatus);
                String msg = "任务状态监听结束，任务状态：" + result.getTaskStatus();
                log.info(msg);
                break;
            }
        }
    }

    //发送状态检查请求
    private CollectorResponse sendStatusCheck(String taskInstanceId) {
        String url = task.getCollectionInterface().getBaseUrl() + "status";

        MultiValueMap<String, String> map = new LinkedMultiValueMap<>();
        map.add("taskInstanceId", taskInstanceId);

        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.MULTIPART_FORM_DATA);
        HttpEntity<Map> entity = new HttpEntity(map, headers);

        CollectorResponse result = (CollectorResponse) remoteRequestService.post(url, entity);
        return result;
    }

    //处理执行日志
    private void handlExecutorLog(String taskInstanceId, CollectorResponse<Integer> result) {
        if (!result.getMessages().isEmpty()) {
            for (String log : result.getMessages()) {
                logWriter.put(ExecuteLog.of(taskInstanceId, log, Calendar.getInstance().getTime()));
            }
        }
        if (!result.getErrors().isEmpty()) {
            for (String log : result.getErrors()) {
                logWriter.put(ExecuteLog.of(taskInstanceId, log, Calendar.getInstance().getTime()));
            }
        }
        if (StringUtils.isNotEmpty(result.getResult())) {
            logWriter.put(ExecuteLog.of(taskInstanceId, result.getResult(), Calendar.getInstance().getTime()));
        }
    }

}
